package cn.ac.iie.di.ban.data.exchange.runner.slave.task.runTask;

import cn.ac.iie.di.ban.data.exchange.runner.commons.protocol.StatusEnum;
import cn.ac.iie.di.ban.data.exchange.runner.commons.worker.SlaveWorker;
import cn.ac.iie.di.ban.data.exchange.runner.slave.BatchSlave;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;

import cn.ac.iie.di.ban.data.exchange.runner.slave.RocketMQHelper.data.RocketMQArgvData;
import cn.ac.iie.di.ban.data.exchange.runner.slave.RocketMQHelper.receiveMQ.ReceiveMQ;
import cn.ac.iie.di.ban.data.exchange.runner.slave.helper.writeFileHelper.WriteFile;
import cn.ac.iie.di.ban.data.exchange.runner.slave.task.checkTask.CheckSQLAndPermissionTask;
import cn.ac.iie.di.commons.util.hdfs.HDFSUtil;
import cn.iie.jsqlparser.helper.JSQLParserHelper;
import cn.iie.jsqlparser.helper.SQLOperatorCompare;
import cn.iie.jsqlparser.helper.data.CondTreeData;
import cn.iie.jsqlparser.helper.data.CondTreeValueData;
import cn.iie.jsqlparser.helper.data.SQLOperatorEnum;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.ParseException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;

import org.apache.commons.net.ftp.FTPClient;

/**
 * Created by Black-Laptop on 12/6/2017.
 */
public class BatchSubscriptionTaskThread extends Thread {

    private static final Logger LOG = LoggerFactory.getLogger(BatchSubscriptionTaskThread.class);

    private Boolean stopThread;
    private RocketMQArgvData rocketMQArgvData;
    private ArrayBlockingQueue<REMessage> queue;
    //    private SendMQ sendMQ;
    private JSQLParserHelper jsqlParserHelper;

    private List<String> selectItems;
    private CondTreeData condTree;

    private String resultPath;

    private long limit;
    private long startTime;
    private String fileName;
    private String sql;

    private String hdfsPath;
    private String srcPath;

    private String ftpIp;
    private String ftpUser;
    private String ftpPassword;
    private int ftpPort;

    public BatchSubscriptionTaskThread(JSQLParserHelper jsqlParserHelper, RocketMQArgvData rocketMQArgvData, String hdfsPath) throws Exception {
        this.stopThread = false;
        this.jsqlParserHelper = jsqlParserHelper;
        this.sql = jsqlParserHelper.getSQLString();
        this.rocketMQArgvData = rocketMQArgvData;
        this.resultPath = resultPath;
        this.startTime = System.currentTimeMillis() / 1000;
        this.fileName = String.valueOf(startTime);
        this.limit = jsqlParserHelper.getLimitValue();
        this.hdfsPath = hdfsPath;
        this.srcPath = "D:\\a\\c\\";
        init();
    }

    @Override
    public void run() {
        LOG.info("sql:" + sql);
        REMessage message = null;
        while (!stopThread) {
            try {
                message = queue.poll(1, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                BatchSlave.callAfterFinished.finished(StatusEnum.BATCH_SUBSCRIPTION_TASK_ERROR, ExceptionUtils.getFullStackTrace(e));
                LOG.error(ExceptionUtils.getFullStackTrace(e));
            }

            if (message != null) {
                getAllowMessage(message);
                REMessage resultMessage = getAllowMessage(message);

                if (resultMessage == null) {
                    continue;
                }

                /**
                 * 以时间差命名文件，limitValue值为间隔时间 每一个limitValue间隔时间创建一个新文件
                 */
                long endTime = System.currentTimeMillis() / 1000;

                LOG.info("start=" + String.valueOf(startTime) + "   endTime=" + String.valueOf(endTime) + "   fileName=" + fileName + "    limit=" + limit);
                if (endTime - startTime >= limit) {
                    startTime = endTime;
                    fileName = String.valueOf(startTime);
                }

                try {
                    writeJsonData(resultMessage);
                } catch (IOException e) {
                    BatchSlave.callAfterFinished.finished(StatusEnum.BATCH_SUBSCRIPTION_TASK_SUCCESS, ExceptionUtils.getFullStackTrace(e));
                    LOG.error(ExceptionUtils.getFullStackTrace(e));
                }

            }
        }
        BatchSlave.callAfterFinished.finished(StatusEnum.BATCH_SUBSCRIPTION_TASK_SUCCESS, sql);
    }

    /**
     * 获得消息队列后以Json格式写入文件
     *
     * @param message
     */
    private void writeJsonData(REMessage message) throws IOException {
        Object[] columns = message.colNames;
        columns = changColumns(columns);
        Object[] types = message.types;
        File srcdirs = new File(srcPath);
        dirExists(srcdirs);
        File srcFile = new File(srcPath + fileName);
        fileExists(srcFile);
        PrintWriter pw = new PrintWriter(new OutputStreamWriter(new FileOutputStream(srcFile), "utf-8"), true);

//        FTPClient ftpClient = new FTPClient();
//        ftpClient.connect(ftpIp,ftpPort); // IP地址
//        ftpClient.login(ftpUser, ftpPassword); // 用户名和密码
//        boolean hasDir = ftpClient.changeWorkingDirectory(resultPath); // 切换到test目录 ,返回boolean值，如果有该文件夹返回true，否则，返回false
//        if (!hasDir) {
//            //创建文件夹
//            ftpClient.makeDirectory(resultPath);
//            ftpClient.changeWorkingDirectory(resultPath);
//        }
//        ftpClient.setControlEncoding("utf-8");
//
//        PrintWriter pw = new  PrintWriter(new OutputStreamWriter(ftpClient.appendFileStream(fileName),"utf-8"),true);
//        WriteFile writeFile = new WriteFile(resultPath, fileName);
        for (Object[] data : message.data) {
//            writeFile.writeFileAppend(getSinaleColumnAndData(columns, data));
            pw.println(getSinaleColumnAndData(columns, data));
            pw.flush();
            pw.close();
        }
        try {
            HDFSUtil.putFileFromLocal(srcPath + fileName, hdfsPath);
            srcFile.delete();
        } catch (Exception ex) {
            LOG.error("upload the operator registration package to HDFS failed. Case:" + ex.getMessage(), ex);
        }

    }

    private void dirExists(File file) {
        if (file.exists()) {
            if (file.isDirectory()) {
//                System.out.println("dir exists");
            } else {
                file.mkdirs();
                System.out.println("same name file exists");
            }
        } else {
//            System.out.println("dir not exists");
            file.mkdirs();
        }
    }

    private void fileExists(File file) {
        if (!file.exists()) {
            try {
                file.createNewFile();
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }

    }

    /**
     * 将发送队列中的列名转换回去
     *
     * @param columns
     * @return
     */
    private Object[] changColumns(Object[] columns) {
//        String postResult = "{\"data\":{\"showchannel\":\"wxshow\",\"srctable\":\"zhj_commentUpdate\",\"srcchannel\":\"wxsrc\",\"cols\":[{\"srccolumn\":\"postchannel\",\"showcolumn\":\"pc\"},{\"srccolumn\":\"postid\",\"showcolumn\":\"pi\"},{\"srccolumn\":\"msg_type\",\"showcolumn\":\"mt\"},{\"srccolumn\":\"domain\",\"showcolumn\":\"d\"},{\"srccolumn\":\"likenum\",\"showcolumn\":\"ln\"},{\"srccolumn\":\"key\",\"showcolumn\":\"k\"}],\"showtable\":\"z_cu\"},\"detail\":\"find msg ok\",\"status\":\"succeeded\"}" ;
        JSONObject jsonPostResult = JSONObject.parseObject(CheckSQLAndPermissionTask.postResult);
        JSONObject jsonDataResult = (JSONObject) jsonPostResult.get("data");
        JSONArray array = jsonDataResult.getJSONArray("cols");
//        System.out.println(array.toString());

        for (int j = 0; j < columns.length; j++) {
            for (int i = 0; i < array.size(); i++) {
                JSONObject jsonColsResult = (JSONObject) array.get(i);
//                System.out.println("cols:"+jsonColsResult.get("srccolumn"));
                if (columns[j].equals(jsonColsResult.get("srccolumn"))) {
                    columns[j] = jsonColsResult.get("showcolumn");
                }
            }
        }
        return columns;
    }

    private String getSinaleColumnAndData(Object[] columns, Object[] data) {
        JSONObject jsonObject = new JSONObject();
        for (int i = 0; i < columns.length; i++) {
            jsonObject.put(String.valueOf(columns[i]), data[i]);
        }
        String result = jsonObject.toString();
        return result;
    }

    public void stopThread() {
        this.stopThread = true;
    }

    private void init() throws Exception {

//        generateTimer.scheduleAtFixedRate(new TokenGenerateTask(), 0, 300*1000);
        queue = new ArrayBlockingQueue<>(5);
        this.selectItems = jsqlParserHelper.getSQLSelectItems();
        this.condTree = jsqlParserHelper.getSQLWhere();

        startReceiveMQ();
    }

    private void startReceiveMQ() throws Exception {
        try {
            ReceiveMQ receiveMQ = new ReceiveMQ(this.rocketMQArgvData, queue);
            receiveMQ.start();
        } catch (REConnectionException | ParseException e) {
            LOG.error(ExceptionUtils.getFullStackTrace(e));
            throw new Exception("start rocketMQ error. ");
        }
    }

    private REMessage getAllowMessage(REMessage message) {
        REMessage allowMessage = null;
        try {
            allowMessage = ScreenAllowMessage(message);
        } catch (Exception e) {
            LOG.error(ExceptionUtils.getFullStackTrace(e));
        }

        if (allowMessage == null || allowMessage.data.size() == 0) {
            return null;
        }

        REMessage resultMessage = getResultMessage(allowMessage);

        try {
        } catch (Exception e) {
            throw new Error(ExceptionUtils.getFullStackTrace(e));
        }

        return resultMessage;
    }

    private REMessage ScreenAllowMessage(REMessage message) throws Exception {
        List<String> colNames = Arrays.asList(message.colNames);
        List<Object[]> datas = message.data;
        List<Object[]> resultDatas = new ArrayList<>();

        for (Object[] data : datas) {
            Boolean dataAllow = checkDataAllow(colNames, data, this.condTree);
            if (dataAllow) {
                resultDatas.add(data);
            }
        }

        if (datas.size() < 1) {
            return null;
        }

        REMessage allowMessage = new REMessage(message.colNames, message.types, resultDatas, message.docSchemaName, message.userDesc);

        return allowMessage;
    }

    private Boolean checkDataAllow(List<String> colNames, Object[] data, CondTreeData whereTree) throws Exception {
        Boolean isLastChild = whereTree.getLeftChild() == null && whereTree.getRightChild() == null;
        if (isLastChild) {
            Boolean compareResult = checkChildDataAllow(colNames, data, whereTree);
            return compareResult;
        }

        Boolean compareResult = checkFatherDataAllow(colNames, data, whereTree);
        return compareResult;
    }

    private Boolean checkChildDataAllow(List<String> colNames, Object[] data, CondTreeData whereTree) throws Exception {
        CondTreeValueData valueData = whereTree.getValue();
        String compareKey = valueData.getKey();
        String compareValue = valueData.getValue();
        SQLOperatorEnum compareOperator = valueData.getOperator();

        int colIndex = colNames.indexOf(compareKey);
        Object value = data[colIndex];

        Boolean compareResult = SQLOperatorCompare.SQLOperatorCompare(compareOperator, value, compareValue);

        return compareResult;
    }

    private Boolean checkFatherDataAllow(List<String> colNames, Object[] data, CondTreeData whereTree) throws Exception {
        CondTreeData leftChild = whereTree.getLeftChild();
        CondTreeData rightChild = whereTree.getRightChild();
        SQLOperatorEnum compareOperator = whereTree.getValue().getOperator();
        Boolean leftChildResult = null;
        Boolean rightChildResult = null;
        if (leftChild != null) {
            leftChildResult = checkDataAllow(colNames, data, leftChild);
        }
        if (rightChild != null) {
            rightChildResult = checkDataAllow(colNames, data, rightChild);
        }

        if (leftChild == null) {
            return rightChildResult;
        } else if (rightChild == null) {
            return leftChildResult;
        }
        Boolean compareResult = SQLOperatorCompare.SQLOperatorCompare(compareOperator, leftChildResult, rightChildResult);

        return compareResult;
    }

    private REMessage getResultMessage(REMessage message) {
        List<Object[]> datas = message.data;
        List<String> colNames = Arrays.asList(message.colNames);
        List<REFieldType> types = Arrays.asList(message.types);

        List<Object[]> datasResult = getResultDatas(message);
        String[] colNamesResult = getResultColumns();
        REFieldType[] typesResult = getResultTypes(message);

        return new REMessage(colNamesResult, typesResult, datasResult, message.docSchemaName, message.userDesc);
    }

    private List<Object[]> getResultDatas(REMessage message) {
        List<Object[]> datas = message.data;
        List<String> colNames = Arrays.asList(message.colNames);

        List<Object[]> datasResult = new ArrayList<>();

        for (Object[] data : datas) {
            List<Object> dataResult = new ArrayList<>();

            for (String column : selectItems) {
                int columnIndex = colNames.indexOf(column);
                dataResult.add(data[columnIndex]);
            }

            datasResult.add(dataResult.toArray(new Object[dataResult.size()]));
        }

        return datasResult;
    }

    private String[] getResultColumns() {
        return selectItems.toArray(new String[selectItems.size()]);
    }

    private REFieldType[] getResultTypes(REMessage message) {
        List<String> colNames = Arrays.asList(message.colNames);
        REFieldType[] types = message.types;

        List<REFieldType> typesResult = new ArrayList<>();

        for (String column : selectItems) {
            int columnIndex = colNames.indexOf(column);
            typesResult.add(types[columnIndex]);
        }

        return typesResult.toArray(new REFieldType[typesResult.size()]);
    }

}
